package cdrprocessing.bolt.filter;

import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class FilterMtcTupleBolt extends BaseBasicBolt {

    public void execute(Tuple input, BasicOutputCollector collector) {
        final String rawCdr = input.getStringByField("rawCdr");
        if (!rawCdr.startsWith("MTC")) {
            collector.emit("RawCdrStream", new Values(rawCdr));
        }
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("RawCdrStream", new Fields("rawCdr"));
    }

}
